#include <iostream>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <memory>
#include "Disruptor/Disruptor.h"
#include "Disruptor/ThreadPerTaskScheduler.h"
#include "Disruptor/YieldingWaitStrategy.h"
#include "Disruptor/BusySpinWaitStrategy.h"
#include "Disruptor/SpinWaitWaitStrategy.h"
#include <sys/time.h>

bool flagProducerOK=false;

using namespace std;
struct timeval starttime,endtime;

struct LongEvent{
    long value;
};

struct PrintingEventHandler : Disruptor::IEventHandler< LongEvent >
{
    explicit PrintingEventHandler(int toProcess) : m_actuallyProcessed(0), m_toProcess(toProcess), sum(0)
    {}

    void onEvent(LongEvent& event, int64_t, bool) override
    {
        
        if (++sum==1){
            std::cout<<"consumer start!"<<std::endl;
        }
        if (++m_actuallyProcessed == m_toProcess){
            std::cout<<"consumer ok!" << std::endl;
            std::cout<<"sum:\t"<<sum<<std::endl;
            gettimeofday(&endtime,NULL);
            std::cout<<"time cost:\t"<<(endtime.tv_sec-starttime.tv_sec)*1000 + (endtime.tv_usec - starttime.tv_usec)/1000 << "ms" << std::endl;
            m_allDone.notify_all();

        }
    }

    void waitEndOfProcessing()
    {
        std::unique_lock<decltype(m_mutex)> lk(m_mutex);
        m_allDone.wait(lk);
    }

private:
    int sum;
    std::mutex m_mutex;
    std::condition_variable m_allDone;
    int m_toProcess;
    int m_actuallyProcessed;
};


int main(){
    auto const ExpectedNumberOfEvents = 40000000;
    auto const RingBufferSize = 67108864;

    // Instantiate and start the disruptor
    auto eventFactory = []() { return LongEvent(); };
    auto taskScheduler = std::make_shared< Disruptor::ThreadPerTaskScheduler >();
    //auto waitStrategy = std::make_shared< Disruptor::YieldingWaitStrategy >();
    //auto waitStrategy = std::make_shared< Disruptor::BusySpinWaitStrategy >();
    auto waitStrategy = std::make_shared< Disruptor::SpinWaitWaitStrategy >();
    
    auto disruptor = std::make_shared< Disruptor::disruptor<LongEvent> >(eventFactory, RingBufferSize, taskScheduler, Disruptor::ProducerType::Single, waitStrategy);
    auto printingEventHandler = std::make_shared< PrintingEventHandler >(ExpectedNumberOfEvents);

    disruptor->handleEventsWith(printingEventHandler);

    taskScheduler->start();
    disruptor->start();

    // Publish events

    gettimeofday(&starttime,NULL);
    auto ringBuffer = disruptor->ringBuffer();
    std::cout<<"producer start!"<<std::endl;
    for (auto i = 0; i<ExpectedNumberOfEvents; ++i)
    {
        auto nextSequence = ringBuffer->next();
        (*ringBuffer)[nextSequence].value = i;
        ringBuffer->publish(nextSequence);
    }
    flagProducerOK=true;
    std::cout<<"producer ok!"<<std::endl;

    // Wait for the end of execution and shutdown
    printingEventHandler->waitEndOfProcessing();

    disruptor->shutdown();
    taskScheduler->stop();



    return 0;
}
